package cn.jly.flink.operators;

import cn.jly.flink.entity.Department;
import cn.jly.flink.entity.Employee;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

/**
 * @PackageName cn.jly.flink.operators
 * @ClassName ConnectDemo
 * @Description 测试connector算子
 * @Author 姬岚洋
 * @Date 2021/1/11 下午 7:14
 */
public class ConnectDemo {
    private static StringBuffer stringBuffer = new StringBuffer();

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();
        env.setParallelism(1);

        final DataStreamSource<Employee> dataStream1 = env.fromElements(
                new Employee(1L, "tom", 1001),
                new Employee(2L, "lanyangji", 1002),
                new Employee(3L, "james", 1001),
                new Employee(4L, "lily", 1002)
        );

        final DataStreamSource<Department> dataStream2 = env.fromElements(
                new Department(1001, "财务部"),
                new Department(1002, "研发部")
        );

        // connect
        final ConnectedStreams<Employee, Department> connect = dataStream1.connect(dataStream2);

        connect.flatMap(new CoFlatMapFunction<Employee, Department, String>() {
            @Override
            public void flatMap1(Employee value, Collector<String> collector) throws Exception {
                stringBuffer.append(value.getName()).append("-");
                collector.collect(stringBuffer.toString());
            }

            @Override
            public void flatMap2(Department value, Collector<String> collector) throws Exception {
                stringBuffer.append(value.getDepartmentName()).append("-");
                collector.collect(stringBuffer.toString());
            }
        })
                /*

                输出：
                    9> lily-
                    11> lily-财务部-
                    12> lily-财务部-研发部-
                    7> lily-财务部-研发部-lanyangji-
                    8> lily-财务部-研发部-lanyangji-james-
                    6> lily-财务部-研发部-lanyangji-james-tom-
                结论：
                    两个流互补干扰，各个元素被处理一次，到达先后不可控
                 */
                .print();


        FlinkUtils.executeStream(env, "ConnectDemo");
    }
}
